/*
 * Copyright 2011 Subhabrata Ghosh <subho.ghosh at langene.net>.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */



package net.langene.common;

//~--- non-JDK imports --------------------------------------------------------

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//~--- JDK imports ------------------------------------------------------------

import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

//~--- classes ----------------------------------------------------------------

/**
 *
 * @author Subhabrata Ghosh <subho.ghosh at langene.net>
 */
public class JobManager implements Runnable {
    private static final Object	_lock    = new Object();
    private static JobManager	_manager = null;
    private final static Logger	_logger  = LoggerFactory.getLogger(JobManager.class);

    //~--- fields -------------------------------------------------------------

    private final HashMap<String, JobHandle>	jobs            = new HashMap<String, JobManager.JobHandle>();
    private int					maxRunningCount = 10;    // Default Max Threads to Run (can be overwritten in config)
    private final Queue<Job<?>>			pending         = new LinkedBlockingQueue<Job<?>>();
    private final Queue<Job<?>>			running         = new LinkedBlockingQueue<Job<?>>();
    private int					sleepIntrvl     = 500;    // Default sleep interval is 500ms (can be overridden in config)
    private boolean				isrunning       = false;

    //~--- constructors -------------------------------------------------------

    private JobManager() throws Exception {
        String	prop = Context.get().getConfig().getString("net.langene.common.jobs.threadcount");

        if ((prop != null) &&!prop.isEmpty()) {
            maxRunningCount = Integer.parseInt(prop);
        }

        prop = Context.get().getConfig().getString("net.langene.common.jobs.sleepinterval");

        if ((prop != null) &&!prop.isEmpty()) {
            sleepIntrvl = Integer.parseInt(prop);
        }
    }

    //~--- methods ------------------------------------------------------------

    private void being() throws Exception {
        new Thread(this).start();
    }

    /**
     * Add the job to the Queue to be scheduled for run.
     *
     * @param job
     * @throws Exception
     */
    public synchronized void add(Job job) throws Exception {
        if (!isrunning) {
            throw new Exception(
                "JobManager is not running or has terminated. Check logs for exceptions");
        }

        synchronized (pending) {
            JobHandle	handle = new JobHandle();

            handle.Job = job;
            jobs.put(handle.Job.id, handle);
            job.status = JobStatus.Pending;
            pending.add(job);
        }
    }

    @Override
    @SuppressWarnings("SleepWhileInLoop")
    public void run() {
        _logger.info("JobManager running...");
        _logger.info("Max Thread Count : " + maxRunningCount);
        isrunning = true;

        try {
            while (true) {
                if (!isrunning) {
                    break;
                }

                if (pending.isEmpty() || (running.size() >= maxRunningCount)) {
                    Thread.sleep(sleepIntrvl);

                    continue;
                }

                for (int ii = 0; ii < (maxRunningCount - running.size()); ii++) {
                    Job<?>	job = null;

                    synchronized (pending) {
                        job = pending.poll();
                    }

                    if (job == null) {
                        break;
                    }

                    if (job.status != JobStatus.Pending) {
                        continue;
                    }

                    synchronized (running) {
                        JobHandle	handle = jobs.get(job.id);
                        Thread		th     = new Thread(job);

                        handle.Thread = th;
                        th.start();
                        running.add(job);
                    }
                }
            }
        } catch (Exception e) {
            _logger.error("Job Manager raised exception...");
            _logger.error("ERROR : " + e.getLocalizedMessage());
            ProgramUtils.printStacktrace(_logger, e);
            isrunning = false;
        }
    }

    /**
     * Mark the current Job as completed and
     * remove it from the Running Queue.
     *
     * @param job
     */
    public void jobFinished(Job<?> job) {
        synchronized (running) {
            running.remove(job);

            JobHandle	handle = jobs.get(job.id);

            handle.Thread = null;
        }
    }

    /**
     * Check if there are Running or Pending Jobs in the queue.
     *
     * @return
     */
    public boolean anyJobRunningOrPending() {
        if ((running.size() > 0) || (pending.size() > 0)) {
            return true;
        }

        return false;
    }

    /**
     * Check if any of the Jobs specified in the List is Running
     * or Pending.
     *
     * @param jobids List of Job ID(s)
     * @return true if any Job is Pending/Running, false if all finished.
     */
    public boolean areJobsRunningOrPending(List<String> jobids) {
        for (String jobid : jobids) {
            Job<?>	job = getJobById(jobid);

            if ((job.status == JobStatus.Running) || (job.status == JobStatus.Pending)) {
                return true;
            }
        }

        return false;
    }

    //~--- get methods --------------------------------------------------------

    /**
     * Get a list of all the Finished/Exception/Terminated jobs
     * from the list of jobids specified.
     *
     * @param jobids List of Job ID(s)
     * @return List of completed Job ID(s)
     */
    public List<String> getFinishedJobs(Collection<String> jobids) {
        List<String>	fjobids = new LinkedList<String>();

        for (String jobid : jobids) {
            Job<?>	job = getJobById(jobid);

            if ((job.status == JobStatus.Finished) || (job.status == JobStatus.Exception)
                    || (job.status == JobStatus.Terminated)) {
                fjobids.add(job.id);
            }
        }

        return fjobids;
    }

    /**
     * Get a Job Object by ID
     *
     * @param jobid
     * @return Job Object if exists, else NULL.
     */
    public Job<?> getJobById(String jobid) {
        if (jobs.containsKey(jobid)) {
            return jobs.get(jobid).Job;
        }

        return null;
    }

    //~--- methods ------------------------------------------------------------

    /**
     * Remove the Job from the Job list;
     * If Running, job will be terminated.
     * If Pending, it will not be scheduled.
     *
     * @param job
     */
    public void remove(Job<?> job) {
        synchronized (pending) {
            if (jobs.containsKey(job.id)) {
                JobHandle	jh = jobs.get(job.id);

                if ((jh.Job.status == JobStatus.Running) || (jh.Job.status == JobStatus.Pending)) {
                    jh.Job.terminate();
                }
            }

            jobs.remove(job.id);
        }
    }

    //~--- get methods --------------------------------------------------------

    /**
     * Get the handle to the JobManager instance.
     * When called for the first time, this method will
     * create a new instance to the JobManager.
     *
     * @return
     * @throws Exception
     */
    @SuppressWarnings("SleepWhileHoldingLock")
    public static JobManager get() throws Exception {
        synchronized (_lock) {
            if (_manager == null) {
                _manager = new JobManager();
                _manager.being();
                Thread.sleep(200);
            }

            return _manager;
        }
    }

    //~--- inner classes ------------------------------------------------------

    private static class JobHandle {

        /** Field description */
        public Job<?>	Job;

        /** Field description */
        public Thread	Thread;
    }
}


//~ Formatted by Jindent --- http://www.jindent.com
